
from datetime import datetime
from kafka import KafkaProducer
import time
import json

# kafka ip端口 修改为你使用的kafka server配置
KAFKA_HOST = "106.120.201.126:19359"    # 公网kafka

# kafka发送方法
# 需要发动到的topic
PRE_DATA_TOPIC = "PRE_DATA_TOPIC1111"
# 发送数据
now_frame_data = {}
producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                         value_serializer=lambda x: json.dumps(x).encode('utf-8'))
count = 0
while True:
    # 事件检测结果
    now_frame_data = [{"test": str(count)}]
    producer.send(PRE_DATA_TOPIC, value=now_frame_data, key="key")
    print("发送数据", now_frame_data)
    count += 1
    time.sleep(0.1 / 10)


# log_path = r'/home/gj/Download/新建文件夹/kako_data_zuduan.log'
# with open(log_path, 'r', encoding='utf-8') as f:
#     current_time = None
#     while True:
#         message = f.readline()
#         if not message:  # 如果文件读取完毕，则退出循环
#             break
#         try:
#             msg = json.loads(message.strip())
#             # 离线数据模拟发送在线数据
#             producer.send(PRE_DATA_TOPIC, value=msg, key="key")
#             print("发送数据", msg)
#             count += 1
#             t = datetime.strptime(msg["timestamp"], "%Y-%m-%dT%H:%M:%S.%f")
#             if current_time is not None:
#                 sleep_duration = (t - current_time).total_seconds()
#                 if sleep_duration > 0:
#                     # print("当前时间:", t, " 跳过:", sleep_duration)
#                     # time.sleep(sleep_duration)
#                     time.sleep(0.2)
#             current_time = t
#
#         except json.JSONDecodeError:
#             print("忽略非JSON格式的日志行")
#             continue